package spider.service;

import com.htiiot.common.db.DbConnectionPool;
import com.htiiot.common.db.JDBCUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import spider.cmd.Cmd;
import spider.config.Configs;
import spider.http.HTTPUtil;
import spider.kafka.SendDataToKafka;
import spider.main.TZDataImport;
import spider.pojo.*;
import spider.util.DateUtil;
import spider.util.JsonUtil;
import spider.util.TidsUtil;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by 李延念 on 2018/5/2.
 */
public class DeviceMetricsService {
    private static Logger logger = LogManager.getLogger(DeviceMetricsService.class);
    private static SendDataToKafka sendDataToKafka;
    private static DbConnectionPool pool;
    private static boolean kafkaFlag = false;
    private static boolean mysqlFlag = false;
    private static String kafkaLastTime = null;
    private static String mysqlLastTime = null;
    static {
        sendDataToKafka = new SendDataToKafka();
        pool  = DbConnectionPool.getPool();
    }

    //接收固定的租户id从表中获取SerialNumbers
    public static List<String> getSerialNumbersByTid(int tid){
        String sql = "SELECT tenant_id ,create_time AS TIME, id,serial_number sn FROM `thing_device` WHERE deleted=0 AND TYPE=0 AND tenant_id in ("+ tid +")";
        logger.info("getSerialNumbersByTid sql : "+ sql);
        Connection conn = pool.getConnection();
        PreparedStatement ps = null;
        ResultSet rs = null;
        List<String> sns = new ArrayList<>();
        try {
            ps = conn.prepareStatement(sql);
            rs = ps.executeQuery();
            while (rs.next()) {
                sns.add(rs.getString("sn"));
            }
        }catch (SQLException e) {
            e.printStackTrace();
        } finally {
            JDBCUtils.close(ps, rs, conn);
        }
        return sns;
    }

    public static Map<String,String> getMetricNameDataStreamName(int tid,String sn){
        String sql = "SELECT td.id id,\n" +
                // "  tc.name metricName,\n" +
                "  tds.name dataStreamName\n" +
                " FROM\n" +
                "  thing_device td,\n" +
                "  thing_component tc,\n" +
                "  thing_data_stream tds,\n" +
                "  thing_device_component_rel tdcr\n" +
                " WHERE td.tenant_id = "+tid+
                "  AND td.serial_number = '"+ sn+"'\n" +
                "  AND td.deleted = 0\n" +
                "  AND td.TYPE = 0\n" +
                "  AND td.id = tdcr.device_id\n" +
                "  AND tdcr.status = 0\n" +
                "  AND tdcr.deleted = 0\n" +
                "  AND tdcr.component_id = tc.id\n" +
                "  AND tdcr.component_id = tds.component_id\n" +
                "  AND tds.status= 0\n" +
                "  AND tds.deleted = 0\n" +
                "  AND tc.status= 0\n" +
                "  AND tc.deleted = 0";
        logger.info("getMetricNameDataStreamName sql : "+ sql);
        Connection conn = pool.getConnection();
        PreparedStatement ps = null;
        ResultSet rs = null;
        List<String> dataStreamNames = new ArrayList<>();
        Map result = new HashMap<String,String>();
        int deviceId = -1;
        try {
            ps = conn.prepareStatement(sql);
            rs = ps.executeQuery();
            while (rs.next()) {
                deviceId = rs.getInt("id");
                dataStreamNames.add(rs.getString("dataStreamName"));
            }
        }catch (SQLException e) {
            e.printStackTrace();
        } finally {
            JDBCUtils.close(ps, rs, conn);
        }

        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < dataStreamNames.size(); i++) {
            String s =  dataStreamNames.get(i);
            sb.append(s);
            if(dataStreamNames.size() >= 0 && i != dataStreamNames.size() -1){
                sb.append(",");
            }
        }
        result.put("deviceId",String.valueOf(deviceId));
        result.put("dataStreamNames",sb.toString());
        logger.info("getMetricNameDataStreamName dataStreamNames : "+ sb.toString());
        return result;
    }

    public static void getMetricDatasFromTZToKafka(){
        List<String> tids = TidsUtil.getTids();
        if(tids == null){
            logger.error( "the tids.properties tids is null" );
            System.exit( 0 );
        }

        for (int k = 0; k < tids.size(); k++) {
            int tid =  Integer.valueOf(tids.get( k ));
            List<String> serialNumbers = getSerialNumbersByTid(tid);
            if(serialNumbers != null && serialNumbers.size() > 0){
                for (int j = 0; j < serialNumbers.size(); j++) {
                    String current = null;
                    String start = null;
                    String sn =  serialNumbers.get(j);
                    Map<String,String> metricNameDataStreamName = getMetricNameDataStreamName(tid, sn);
                    List<String> betweenDates = null;
                    List<String> startDates = new ArrayList<>();
                    List<String> endDates = new ArrayList<>();
                    if(!kafkaFlag){
                        //启动,有开始，从开始到此刻
                        if(Cmd.startTime != null){
                            betweenDates = DateUtil.getBetweenDates( Cmd.startTime, DateUtil.getyyyyMMdd(-1));
                            for(String date : betweenDates){
                                startDates.add( date + " 00:00:00" );
                                endDates.add( date + " 23:59:59" );
                            }
                        }
                        start = DateUtil.getStart();      //2018-07-02 00:00:00
                        current = DateUtil.getCurrent(); //2018-07-02 16:12:04
                        startDates.add( start);
                        endDates.add( current);
                        if(j == serialNumbers.size() -1){
                            kafkaFlag = true;
                        }
                    }else{
                        start = kafkaLastTime;
                        current = DateUtil.getCurrent();
                        //大于0表示跨天
                        if(DateUtil.getStep( start,current ) > 0){
                            try {
                                startDates.add( start);
                                endDates.add( DateUtil.getEnd( start ));
                                startDates.add( DateUtil.getStart( current ));
                                endDates.add( current);
                            } catch (ParseException e) {
                                e.printStackTrace();
                            }
                        }else {
                            startDates.add( start );
                            endDates.add( current );
                        }
                    }
                    logger.info("getMetricDatasFromTZToKafka start time is " + startDates.toString() + " and end time is "+ endDates.toString());
                    if(j == serialNumbers.size() -1){
                        kafkaLastTime = current;
                        logger.info( "getMetricDatasFromTZToKafka kafkaLastTime is " + kafkaLastTime );
                    }
                    //从天正接口获取数据
                    int did = Integer.valueOf(metricNameDataStreamName.get("deviceId"));
                    String dataStreamNames = metricNameDataStreamName.get("dataStreamNames");
                    int size = startDates.size();
                    for(int i = 0; i < size; i++){
                        String startTime = startDates.get( i ).replace(" ","%20");
                        String endTime = endDates.get( i ).replace(" ","%20");
                        //从天正接口获取数
                       getDateFromTZSendDataToKafka(did,tid,dataStreamNames,sn,startTime,endTime);
                    }
                }
            }
        }

    }

    private static void getDateFromTZSendDataToKafka(int did, int tid, String dataStreamNames, String sn, String startTime, String endTime) {
        String url = "http://outwardApi.hence.cn/api/datas/getList?Token=d04259b9c0a3ccabedae048c1d3bef3f&dataStreamName="+dataStreamNames+"&deviceId="+sn+"&start="+startTime+"&end="+endTime;
        logger.info("getMetricDatasFromTZToKafkaCopy url is : "+url);
        List<DatasTotal> datasTotals = HTTPUtil.get(url, DatasTotal.class);
        if(datasTotals != null && datasTotals.size() > 0){
            DatasTotal datasTotal = datasTotals.get(0);
            String code = datasTotal.getCode();
            if("0000".equals(code)){
                Datass data = datasTotal.getData();
                if(data != null){
                    List<DeviceMetric> datas = data.getDatas();
                    if(datas != null && datas.size() > 0){
                        for (int i = 0; i < datas.size(); i++) {
                            DeviceMetric deviceMetric =  datas.get(i);
                            //String company = deviceMetric.getCompany();
                            String dataStreamName = deviceMetric.getDataStreamName();
                            if("Status".equals(dataStreamName)){
                                //跳过状态数据,状态数据入另外topic
                                logger.info("this dataStream Name is Status");
                                continue;
                            }

                            StreamingData streamingData = new StreamingData();
                            streamingData.setDid(did);
                            streamingData.setTs(System.currentTimeMillis());
                            streamingData.setTid(tid);
                            List<MetricValue> values = deviceMetric.getValues();
                            if(values != null && values.size() > 0){
                                List<StreamingMetricData> mList = new ArrayList<StreamingMetricData>();
                                for (int k = 0; k < values.size(); k++) {
                                    StreamingMetricData metricData = new StreamingMetricData();
                                    metricData.setK(dataStreamName);
                                    //测点id
                                    int metricId = getMetricId(dataStreamName, tid, did);
                                    logger.info("dataStreamName, tid, deviceId "+ dataStreamName+","+tid+","+ did);
                                    metricData.setDn(new DeviceNumber(tid,did,metricId).toHexString());
                                    MetricValue metricValue =  values.get(k);
                                    Long ts = Long.valueOf(metricValue.getTs());
                                    metricData.setTs(ts);
                                    String value = metricValue.getValue();
                                            /*if("Status".equals(dataStreamName)){
                                                metricData.setK("status");
//                                                if("0".equals(value)){
//                                                    value = "16";
//                                                }else if("1".equals(value)){
//                                                    value = "64";
//                                                }else if("2".equals(value)){
//                                                    value = "0";
//                                                }
                                                if("0".equals(value)){
                                                    value = "1000";
                                                }else if("1".equals(value)){
                                                    value = "2000";
                                                }else if("2".equals(value)){
                                                    value = "4000";
                                                }
                                            }*/
                                    metricData.setV(value);
                                    mList.add(metricData);
                                }
                                //过滤datas为空
                                if(mList != null && mList.size() > 0){
                                    streamingData.setData(mList);
                                }
                            }
                            if(streamingData.getData() != null && streamingData.getData().size() > 0){
                                //当有数据时才发送到kafka
                                sendDataToKafka.send(Configs.KafkaConfig.TOPIC, JsonUtil.buildResult(streamingData));
                            }
                        }
                    }
                }
            }
        }
    }

    public static void getMetricStatussFromTZToMysql() {
        List<String> tids = TidsUtil.getTids();
        if(tids == null){
            logger.error( "the tids.properties tids is null" );
            System.exit( 0 );
        }

        for (int k = 0; k < tids.size(); k++) {
            int tid = Integer.valueOf( tids.get( k ) );
            List<String> serialNumbers = getSerialNumbersByTid(tid);
            if(serialNumbers != null && serialNumbers.size() > 0){
                for (int j = 0; j < serialNumbers.size(); j++) {
                    String start = null;
                    String current = null;
                    String sn =  serialNumbers.get(j);
                    Map<String,String> metricNameDataStreamName = getMetricNameDataStreamName(tid, sn);
                    List<String> betweenDates = null;
                    List<String> startDates = new ArrayList<>();
                    List<String> endDates = new ArrayList<>();
                    if(!mysqlFlag){
                        //启动,有开始，从开始到此刻
                        if(Cmd.startTime != null){
                            betweenDates = DateUtil.getBetweenDates( Cmd.startTime, DateUtil.getyyyyMMdd(-1));
                            for(String date : betweenDates){
                                startDates.add( date + " 00:00:00" );
                                endDates.add( date + " 23:59:59" );
                            }
                        }
                        start = DateUtil.getStart();      //2018-07-02 00:00:00
                        current = DateUtil.getCurrent(); //2018-07-02 16:12:04
                        startDates.add( start);
                        endDates.add( current);
                        if(j == serialNumbers.size() -1){
                            mysqlFlag = true;
                        }
                    }else{
                        start = mysqlLastTime;
                        current = DateUtil.getCurrent();
                        //大于0表示跨天
                        if(DateUtil.getStep( start,current ) > 0){
                            try {
                                startDates.add( start);
                                endDates.add( DateUtil.getEnd( start ));
                                startDates.add( DateUtil.getStart( current ));
                                endDates.add( current);
                            } catch (ParseException e) {
                                e.printStackTrace();
                            }
                        }else {
                            startDates.add( start );
                            endDates.add( current );
                        }
                    }
                    logger.info("getMetricDatasFromTZToKafka start time is " + startDates.toString() + " and end time is "+ endDates.toString());
                    if(j == serialNumbers.size() -1){
                        mysqlLastTime = current;
                        logger.info( "getMetricDatasFromTZToKafka kafkaLastTime is " + mysqlLastTime );
                    }
                    int did = Integer.valueOf(metricNameDataStreamName.get("deviceId"));
                    //从天正接口获取数据
                    //从天正接口获取数据
                    int size = startDates.size();
                    for(int i = 0; i < size; i++){
                        String startTime = startDates.get( i ).replace(" ","%20");
                        String endTime = endDates.get( i ).replace(" ","%20");
                        //从天正接口获取数
                        getDataFromTZSendToMysql(did,sn,startTime,endTime);
                    }
                }
            }
        }
    }

    private static void getDataFromTZSendToMysql(int did, String sn, String startTime, String endTime) {
        String url = "http://outwardApi.hence.cn/api/datas/getList?Token=d04259b9c0a3ccabedae048c1d3bef3f&dataStreamName=Status&deviceId="+sn+"&start="+startTime+"&end="+endTime;
        logger.info("getMetricStatussFromTZToMysql url is : "+url);
        List<DatasTotal> datasTotals = HTTPUtil.get(url, DatasTotal.class);
        if(datasTotals != null && datasTotals.size() > 0){
            DatasTotal datasTotal = datasTotals.get(0);
            String code = datasTotal.getCode();
            if("0000".equals(code)){
                Connection conn = pool.getConnection();
                PreparedStatement ps = null;
                ResultSet rs = null;
                try {
                    conn.setAutoCommit(false);
                    String sql = "insert into tz_device_state_calc(device_id,ts,value) values (?,?,?)";
                    ps = conn.prepareStatement(sql);
                    ps.clearBatch();
                    Datass data = datasTotal.getData();
                    if(data != null){
                        List<DeviceMetric> datas = data.getDatas();
                        if(datas != null && datas.size() > 0){
                            for (int i = 0; i < datas.size(); i++) {
                                logger.info("getDataFromTZSendToMysql is "+ datas);
                                DeviceMetric deviceMetric =  datas.get(i);
                                List<MetricValue> values = deviceMetric.getValues();
                                if(values != null && values.size() > 0){
                                    for (int k = 0; k < values.size(); k++) {
                                        MetricValue metricValue = values.get(k);
                                        long ts = Long.valueOf(metricValue.getTs());
                                        ps.setInt(1,did);
                                        ps.setLong(2,ts);
                                        ps.setInt(3,Integer.valueOf(metricValue.getValue()));
                                        ps.addBatch();
                                    }
                                }
                            }
                            ps.executeBatch();
                        }
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                    try {
                        conn.rollback();
                    } catch (SQLException e1) {
                        e1.printStackTrace();
                    }
                } finally {
                    try {
                        conn.commit();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                    JDBCUtils.close(ps, rs, conn);
                }
            }
        }
    }


    public static int getMetricId(String metricName,int tenantId,int deviceId){
        String key = metricName+"_"+tenantId+"_"+deviceId;
        if(TZDataImport.companyDeviceMetricId.containsKey(key)){
            return TZDataImport.companyDeviceMetricId.get(key);
        }
        String sql = "SELECT r.id id FROM thing_device_component_rel r ,thing_data_stream s WHERE r.component_id = s.component_id AND s.name = '"+metricName+"' AND s.tenant_id = "+ tenantId +" AND r.device_id = "+deviceId;
        Connection conn = pool.getConnection();
        PreparedStatement ps = null;
        ResultSet rs = null;
        int metricId = -1;
        try {
            ps = conn.prepareStatement(sql);
            rs = ps.executeQuery();
            while (rs.next()) {
                metricId = rs.getInt("id");
                TZDataImport.companyDeviceMetricId.put(key,metricId);
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }finally {
            JDBCUtils.close(ps, rs, conn);
        }
        return metricId;
    }
}
